package com.bj58.mis.apm.collector.kylog;

import com.bj58.mis.apm.client.core.AppNodeContext;
import com.bj58.mis.apm.collector.BaseSchedulableCollector;
import org.hyperic.sigar.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.Reader;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Deque;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class KylinLogCollector 	extends BaseSchedulableCollector<KyLogList> {
    protected Logger logger = LoggerFactory.getLogger(this.getClass());
    private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private Pattern groupParrern = Pattern.compile("groupsD=\\[.*?\\]");
    private Pattern filterPattern = Pattern.compile("filterD=\\[.*?\\]");
    private Deque<KyLog> queue=new ConcurrentLinkedDeque<KyLog>();
    private Map<String,KyLog> map=new ConcurrentHashMap<String,KyLog>();
    private String contain = AppNodeContext.getValue("kylin.query.log.sql.contain");
    private String notContain = AppNodeContext.getValue("kylin.query.log.sql.notcontain");
    private String[] segmentPattern = {"The scan "," for segment "," is as below with "};
    public KylinLogCollector() {

        String kyLogFile= AppNodeContext.getValue("kylin.log");
        //String kyLogFile="D:\\test.log";
        Sigar sigar = new Sigar();

        FileWatcherThread watcherThread =
                FileWatcherThread.getInstance();

        watcherThread.doStart();

        watcherThread.setInterval(1000);

        FileTail watcher =
                new FileTail(sigar) {
                    public void tail(FileInfo info, Reader reader) {
                        String line;
                        BufferedReader buffer =
                                new BufferedReader(reader);

                        if (getFiles().size() > 1) {
                            logger.info("==> " +
                                    info.getName() +
                                    " <==");
                        }

                        try {
                            final String sepLine="==========================[QUERY]===============================";

                            String prevQueryId="";
                            String prevLine="";
                            while ((line = buffer.readLine()) != null) {
                                try {
                                    String queryId="";
                                    /*
                                        如下判断是为了提取 本行应有的queryId,分为4种情况.
                                        如果上一行是分隔符行,并且本行包括 Query Id 则表示这是一个统计块的开始.这种情况则给prevQueryId 赋值
                                        如果上一行是分隔符行,并且本行不包括 Query Id 则表示这是统计块的结束.清空prevQueryId
                                        如果是统计块的其他行,则继承prevQueryId 为本行的queryId
                                        如果不是统计块,则通过行内容截取出queryId
                                     */

                                    if(sepLine.equals(prevLine)&&line.contains("Query Id:")){
                                        prevQueryId=line.replaceAll("Query Id: ","").trim();
                                        queryId=prevQueryId;
                                    }
                                    else if(sepLine.equals(prevLine)&& !line.contains("Query Id:")){
                                        KyLog log=map.remove(prevQueryId);
                                        if(log!=null && checkSqlContain(log.getSql())) {
                                            queue.offer(log);
                                        }
                                        prevQueryId="";
                                        queryId="";
                                    }
                                    else if(!"".equals(prevQueryId)){
                                        queryId=prevQueryId;
                                    }else {
                                        //如果是新的一个查询,获取QueryID
                                        queryId=getQueryID(line);
                                    }

                                    if(!"".equals(queryId)){
                                        parseString(line,queryId);
                                    }
                                    prevLine=line;
                                } catch (ParseException e) {
                                    logger.error(e.getMessage(),e);
                                }
                            }
                        } catch (IOException e) {
                            logger.error(e.getMessage(),e);
                        }
                    }
                };

        try {
            watcher.add(kyLogFile);
        } catch (SigarException e) {
            logger.error(e.getMessage(),e);
        }


        watcherThread.add(watcher);

    }

    private String getQueryID(String line){

        String key="[Query ";
        if(line.contains(key)) {
            int startIndex = line.indexOf(key) + key.length();
            int endIndex = line.lastIndexOf("-");
            if(startIndex!=-1 && endIndex!=-1) {
                try {
                    String queryId = line.substring(startIndex, endIndex);
                    return queryId;
                } catch(StringIndexOutOfBoundsException ex){
                    //substring解析异常则返回空串
                    return "";
                }
            }
            //如果不是以 "[Query " 开头, 并存在 "-" 则返回空串
            return "";
        }
        //不包含 "[Query " 则返回空串
        return "";
    }

    private void parseString( String line,String queryId) throws ParseException {


         //如果是新的查询,则创建新日志对象
        if(line.contains("[Query") && line.contains("Using project:")){

            KyLog log=new KyLog();
            log.setQueryId(queryId);
            log.setQueryTime(format.parse(line.substring(0,18)));
            map.put(queryId,log);
        }

        /**
         * 判断两种情况.
         * 第一种情况是普通的日志行
         * 第二种情况是汇总的统计行
         * 两种情况各自解析不同的内容
         */
        if(line.contains("[Query")&&!"".equals(queryId)){
            KyLog log =map.get(queryId);
            if(log !=null){
                //处理groupsD和filterD
                if(line.contains("groupsD") && line.contains("filterD")){
                    Matcher m=groupParrern.matcher(line);
                    if(m.find()){
                        StringBuilder sb=new StringBuilder();
                        for(String str:m.group().replaceAll("(?:groupsD=\\[|\\])","").split(",")){
                            sb.append(str.trim());
                            sb.append(",");
                        }
                        log.setGroupList(sb.toString());
                    }
                    m=filterPattern.matcher(line);
                    if(m.find()){
                        StringBuilder sb=new StringBuilder();
                        for(String str:m.group().replaceAll("(?:filterD=\\[|\\])","").split(",")){
                            sb.append(str.trim());
                            sb.append(",");
                        }
                        log.setFilterList(sb.toString());
                    }
                }

                /*
                if(line.contains("Init CubeSegmentScanner for segment")){
                    String str=line;
                    String key="Init CubeSegmentScanner for segment";
                    str=str.substring(str.indexOf(key)).replaceAll("Init CubeSegmentScanner for segment","").trim();
                    log.setSegment(str);
                }
                */
                dealSegment(log, line);
            }
        }
        else if (!line.contains("[Query")&&!"".equals(queryId)){
            KyLog log=map.get(queryId);
            if(log!=null) {
                if (line.contains("SQL:")) {
                    log.setSql(line.replaceAll("SQL:", "").trim());
                }
                /**
                 *  多行SQL合并为一行
                 *  如果log.getSql()没有最后的换行符,则说明没有结束.
                 *  在User段,会补齐log.getSql()的换行符
                 */
                if(!log.getSql().endsWith("\n") && !line.contains("SQL:") && !line.contains("User:")){
                    String sql=log.getSql();
                    log.setSql(sql+"\n"+line);
                }

                if (line.contains("User:")) {
                    log.setUser(line.replaceAll("User:", "").trim());
                    log.setSql(log.getSql()+"\n");
                }
                if (line.contains("Success:")) {
                    log.setSuccess(line.replaceAll("Success:", "").trim());
                }
                if (line.contains("Duration:")) {
                    log.setDuration(line.replaceAll("Duration:", "").trim());
                }
                if (line.contains("Project:")) {
                    log.setProject(line.replaceAll("Project:", "").trim());
                }
                if (line.contains("Realization Names:")) {
                    log.setRealizationNames(line.replaceAll("Realization Names:", "").trim());
                }
                if (line.contains("Cuboid Ids:")) {
                    log.setCuboidIds(line.replaceAll("Cuboid Ids:", "").trim());
                }
                if (line.contains("Total scan count:")) {
                    log.setTotalScanCount(Integer.valueOf(line.replaceAll("Total scan count:", "").trim()));
                }
                if (line.contains("Total scan bytes:")) {
                    log.setTotalScanBytes(Integer.valueOf(line.replaceAll("Total scan bytes:", "").trim()));
                }
                if (line.contains("Result row count:")) {
                    log.setResultRowCount(Integer.valueOf(line.replaceAll("Result row count:", "").trim()));
                }
                if (line.contains("Accept Partial:")) {
                    if (line.contains("false")) {
                        log.setAcceptPartial(false);
                    } else {
                        log.setAcceptPartial(true);
                    }
                }
                if (line.contains("Is Partial Result:")) {
                    if (line.contains("false")) {
                        log.setPartialResult(false);
                    } else {
                        log.setPartialResult(true);
                    }
                }
                if (line.contains("Hit Exception Cache:")) {
                    if (line.contains("false")) {
                        log.setHitExceptionCache(false);
                    } else {
                        log.setHitExceptionCache(true);
                    }
                }
                if (line.contains("Storage cache used:")) {
                    if (line.contains("false")) {
                        log.setStorageCacheUsed(false);
                    } else {
                        log.setStorageCacheUsed(true);
                    }
                }
                if (line.contains("Is Query Push-Down:")) {
                    if (line.contains("false")) {
                        log.setQueryPushDown(false);
                    } else {
                        log.setQueryPushDown(true);
                    }
                }
                if (line.contains("Is Prepare:")) {
                    if (line.contains("false")) {
                        log.setPrepare(false);
                    } else {
                        log.setPrepare(true);
                    }
                }
                if (line.contains("Trace URL:")) {
                    log.setTraceURL(line.replaceAll("(Trace URL:|null)", "").trim());
                }


                if (line.contains("Message:")) {
                    log.setMessage(line.replaceAll("(Message:|null)", "").trim());
                }
            }
        }

        //把原始日志保存到对象
        KyLog log=map.get(queryId);
        if(log!=null){
            log.getSb().append(line+"\n");
        }
    }

    @Override
    public void schedule() {
        collector();
    }

    public void collector() {
        KyLogList list=new KyLogList();
        int max=100;
        KyLog log=null;
        while(max>0 && queue.size()>0) {
            log = queue.removeFirst();
            max--;
            list.getLogList().add(log);
        }
        logger.debug("record size:{} record:{}",list.getLogList().size(),list.toString());
        if(list.getLogList().size()>0){
            super.collect(list);
            if(queue.size() > 0){
                collector();
            }
        }
    }

    private boolean checkSqlContain(String sql){
        sql = sql.toLowerCase();
        if(this.contain != null && !"".equals(contain)) {
            String[] contains = contain.split(",");
            for (String con : contains) {
                if (sql.indexOf(con.toLowerCase()) == -1) {
                    return false;
                }
            }
        }
        if(this.contain != null && !"".equals(notContain)) {
            String[] notContains = notContain.split(",");
            for (String notcon : notContains) {
                if (sql.startsWith(notcon.toLowerCase())){
                    return false;
                }
            }
        }
        return true;
    }

    private void dealSegment(KyLog log, String line){
        if(line.indexOf(segmentPattern[0]) > 0 && line.indexOf(segmentPattern[1]) > 0 && line.indexOf(segmentPattern[2]) > 0){
            String currentSegment = line.substring(line.indexOf(segmentPattern[1])+segmentPattern[1].length(),line.indexOf(segmentPattern[2]));
            if(currentSegment != null && currentSegment.length() > 0 && currentSegment.indexOf("[") > 0 && currentSegment.indexOf("]") > 0){
                String segment = log.getSegment();
                if(segment == null || "".equals(segment)){
                    segment = currentSegment;
                }else{
                    segment = segment + "," + currentSegment;
                }
                log.setSegment(segment);
            }
        }
    }

}